from pyspark.sql import SparkSession
import os
import threading

os.environ['JAVA_HOME'] = '/home/xiaodd/jdk8'  # 这里的路径为java的bin目录所在路径
import globalConfig
import pymysql
from DBUtils.PooledDB import PooledDB
if globalConfig.db == "spark":
    sp = SparkSession.builder.appName("test").config("spark.sql.warehouse.dir",
                                                     "hdfs://localhost:9000/user/hive/warehouse") \
        .config("spark.hadoop.hive.metastore.uris", "thrift://localhost:9083").enableHiveSupport().getOrCreate()

if globalConfig.db == "mysql":
    pool = PooledDB(creator=pymysql, mincached=10, host=globalConfig.db_ip, user='root', passwd='root', db='bdmo',
                    port=3306, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor)


def execute_a_sql(sql, type="select"):
    if globalConfig.db == "spark":
        result = sp.sql(sql)
        result = result.toPandas().to_dict("records")
        return result
    if globalConfig.db == "mysql":
        conn = pool.connection(shareable=False)
        cur = conn.cursor()
        cur.execute(sql)
        if type == "select":
            result = cur.fetchall()
        else:
            result = "success"
        conn.commit()
        cur.close()
        conn.close()
        return result


def create_index_table():
    if globalConfig.db == "mysql":
        type = "varchar(50)"
    else:
        type = "string"
    sql = "create table tag_item(tag_id " + type + ", item_id " + type + ")"
    execute_a_sql(sql, "change")


def create_item_table(table_name, cols):
    if globalConfig.db == "mysql":
        type = "varchar(50)"
    else:
        type = "string"

    sql = "create table " + table_name + "("
    for cur_col in cols[:-1]:
        sql = sql + cur_col + " " + type + ", "
    sql = sql + cols[-1] + " " + type + ")"
    print(sql)
    execute_a_sql(sql, "change")


def insert_item_table(table_name, values):
    sql = "insert into " + table_name + " values("
    for value in values[:-1]:
        sql = sql + '"' + value + '"' + ", "
    sql = sql + '"' + values[-1] + '"' + ")"
    print(sql)
    execute_a_sql(sql, "change")


def insert_index_table(tag_id, item_id):
    sql = "insert into tag_item values(" + tag_id + "," + item_id + ")"
    execute_a_sql(sql, "change")


def delete_index_table(tag_id, item_id):
    sql = "delete from tag_item where tag_id=" + tag_id + " and item_id=" + item_id
    execute_a_sql(sql, "change")


# the return format is ['id1', 'id2' ...]
def get_tag_ids_by_item_id(item_id):
    sql = "select tag_id from tag_item where item_id=" + "\"" + item_id + "\""
    re = execute_a_sql(sql)
    result = []
    for cur in re:
        result.append(cur["tag_id"])
    return result


def get_tag_count_by_item_ids(item_ids):
    sql = "select tag_id, count(*) as num from tag_item where item_id in ("
    sql = sql + item_ids[0]
    for cur_id in item_ids[1:]:
        sql = sql + "," + cur_id
    sql = sql + ") group by tag_id;"
    print(sql)
    re = execute_a_sql(sql)
    print(re)
    return re


# the return format is ['id1', 'id2' ...]
def get_item_ids_by_tag_id(tag_id):
    sql = "select item_id from tag_item where tag_id=" + "\"" + tag_id + "\""
    print(sql)
    result = []
    for cur in execute_a_sql(sql):
        result.append(cur['item_id'])
    return result


def get_item_id_by_condition(item_table_name, tag_json):
    id = tag_json['id']
    sql = "select item_id from " + item_table_name + " where"
    # {"property": "age", "type": "value", "content": ["18", "-"]},
    # {"property": "type", "type": "enum", "content": ["typeA", "typeB"]}
    is_first = True
    for cur_condition in tag_json['conditions']:
        if is_first:
            pre_fix = " "
            is_first = False
        else:
            pre_fix = " and "
        if cur_condition['type'] == "value":
            if cur_condition['content'][0] == "-":
                sql = sql + pre_fix + cur_condition['property'] + " < " + cur_condition['content'][1]
            elif cur_condition['content'][1] == "-":
                sql = sql + pre_fix + cur_condition['property'] + " > " + cur_condition['content'][0]
            else:
                sql = sql + pre_fix + cur_condition['property'] + " between " + \
                      cur_condition['content'][0] + " and " + cur_condition['content'][1]
        else:
            sql = sql + pre_fix + cur_condition['property'] + " in " + "("
            for enu in cur_condition['content']:
                sql = sql + "\"" + enu + '",'
            sql = sql[0:len(sql) - 1]
            sql = sql + ")"
    sql = sql + ";"
    print(sql)
    return execute_a_sql(sql)


def add_tag_to_item(item_table_name, tag_json):
    ids = get_item_id_by_condition(item_table_name, tag_json)
    tag_id = tag_json['id']
    print(tag_id)
    for item_id in ids:
        # print(item_id["item_id"])
        insert_index_table(tag_id, item_id['item_id'])
    return "success"


def get_group_by_tags(tag_ids):
    result = get_item_ids_by_tag_id(tag_ids[0])
    # print(result)
    for cur_tag_id in tag_ids[1:]:
        cur = get_item_ids_by_tag_id(cur_tag_id)
        result = [v for v in result if v in cur]
    # print("here")
    # print(result)
    return result


def get_item_info_by_ids_and_cols(ids, cols, topic):
    sql = "select " + cols[0]
    for cur_col in cols[1:]:
        sql = sql + "," + cur_col
    sql = sql + " from " + topic
    sql = sql + " where item_id in("
    sql = sql + "\"" + ids[0] + "\""
    for cur_id in ids:
        sql = sql + ",\"" + cur_id + "\""
    sql = sql + ");"
    print(sql)
    result = execute_a_sql(sql)
    return result


def old_get_tags_distribution_by_tag_id(tag_ids):
    print(tag_ids)
    print(type(tag_ids))
    item_ids = get_group_by_tags(tag_ids)
    result = {}
    for item_id in item_ids:
        item_tag_ids = get_tag_ids_by_item_id(item_id)
        for item_tag_id in item_tag_ids:
            if item_tag_id not in result.keys():
                result[item_tag_id] = 1
            else:
                result[item_tag_id] += 1
    print(result)
    return result


def get_tags_distribution_by_tag_id(tag_ids):
    print(tag_ids)
    print(type(tag_ids))
    item_ids = get_group_by_tags(tag_ids)
    result = {}
    step = 10
    item_id_list = [item_ids[i:i + step] for i in range(0, len(item_ids), step)]
    # print(item_id_list)
    for cur_item_ids in item_id_list:
        cur_re = get_tag_count_by_item_ids(cur_item_ids)
        for id_nums in cur_re:
            tag_id = id_nums['tag_id']
            num = id_nums['num']
            if tag_id not in result.keys():
                result[tag_id] = 1
            else:
                result[tag_id] += num
    print(result)
    return result


if __name__ == "__main__":
    # init the index table
    # data = {
    #     "id": 123,
    #     "name": "test_json",
    #     "topic": "test_topic",
    #     "conditions": [
    #         {"property": "age", "type":  "value", "content": ["18", "40"]},
    #         {"property": "type", "type":  "enum", "content":  ["typeA", "typeB"]}
    #     ]
    # }
    # create_index_table()
    # item_ids = [i for i in range(102)]
    # step = 1000
    # item_id_list = [item_ids[i:i + step] for i in range(0, len(item_ids), step)]
    # print(item_id_list)
    re = execute_a_sql("select * from antipoor where item_id=2;")
    print(re)